package com.iakuil.app.common.tp;

import cn.hutool.core.thread.ExecutorBuilder;
import cn.hutool.core.thread.RejectPolicy;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * 线程池配置
 *
 * @author Kai
 */
@Slf4j
@EnableAsync(proxyTargetClass = true)
@AutoConfiguration
@EnableConfigurationProperties(ThreadPoolProperties.class)
@Import(CustomThreadPoolRegistrar.class)
public class ThreadPoolAutoConfiguration implements AsyncConfigurer {

    /**
     * 默认阻塞队列容量
     */
    private static final int QUEUE_CAPACITY = 128;

    /**
     * CPU核心数
     *
     * <p>极个别情况，比如虚拟机，该方法获取的CPU核心数不准确会造成启动失败。
     */
    private static final int CPU_PROCESSORS_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * 默认连接池
     *
     * <p>用于CPU密集型任务。
     */
    @Bean("defaultTaskExecutor")
    public Executor defaultTaskExecutor() {
        int threadSize = CPU_PROCESSORS_COUNT + 1;
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(threadSize);
        executor.setMaxPoolSize(threadSize << 1);
        executor.setQueueCapacity(QUEUE_CAPACITY);
        executor.setThreadNamePrefix("default-thread-pool");
        executor.initialize();
        return executor;
    }

    /**
     * 指定Spring默认线程池
     */
    @Override
    public Executor getAsyncExecutor() {
        return defaultTaskExecutor();
    }

    /**
     * 默认线程池异常处理
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return ((throwable, method, objects) -> log.error("Occurring an exception during async task [{}] invocation!\n{}", method.getName(), throwable.getMessage()));
    }

    // ================================ 自定义线程池，使用Hutool封装 =================================

    /**
     * 业务操作日志连接池
     *
     * <p>日志属于典型的IO密集型任务。
     */
    @Bean("operateLogExecutor")
    public Executor operateLogExecutor() {
        int threadSize = CPU_PROCESSORS_COUNT * 2;
        return ExecutorBuilder.create()
            .setCorePoolSize(threadSize)
            .setMaxPoolSize(threadSize << 1)
            .useArrayBlockingQueue(QUEUE_CAPACITY)
            .setKeepAliveTime(60, TimeUnit.SECONDS)
            .setHandler(RejectPolicy.DISCARD.getValue())
            .setThreadFactory(ThreadFactoryBuilder.create().setNamePrefix("oplog-thread-pool-").build())
            .build();
    }

    /**
     * Redis Stream消息轮询连接池
     */
    @Bean("redisListenerPollingExecutor")
    public Executor redisListenerPollingExecutor() {
        return ExecutorBuilder.create()
            .setCorePoolSize(4)
            .setMaxPoolSize(4 << 1)
            .useArrayBlockingQueue(QUEUE_CAPACITY)
            .setKeepAliveTime(30, TimeUnit.SECONDS)
            .setHandler(RejectPolicy.ABORT.getValue())
            .setThreadFactory(ThreadFactoryBuilder.create().setNamePrefix("redis-polling-pool-").build())
            .build();
    }
}
